package com.chinamobile.websocket.websocketdispatcher.distributor.kafka;

import com.alibaba.fastjson.JSON;
import com.chinamobile.websocket.domain.bo.WebsocketKafkaMessageBo;
import com.chinamobile.websocket.domain.dto.MessageDto;
import com.chinamobile.websocket.websocketdispatcher.common.MessageDistributorConstants;
import com.chinamobile.websocket.websocketdispatcher.distributor.MessageDistributor;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;

/**
 * @className: WebsocketListening
 * @description 订阅队列监听
 * @author: fanzhijun
 * @date: 2024/1/24
 **/
@Data
@Slf4j
public class WebsocketListening {
    private MessageDistributor<MessageDto> messageDistributor;

    //监听websocket订阅消费队列
    @KafkaListener(topics = MessageDistributorConstants.KAFKA_WEBSOCKET_MESSAGE_TOPIC, topicPartitions = {@TopicPartition(topic = MessageDistributorConstants.KAFKA_WEBSOCKET_MESSAGE_TOPIC, partitions = "0")})
    public void websocketListening(List<ConsumerRecord<String, WebsocketKafkaMessageBo>> messageRecords, Acknowledgment acknowledgment) {
        log.debug("kafka websocketListening...");
        for (ConsumerRecord<String, WebsocketKafkaMessageBo> record : messageRecords) {
            WebsocketKafkaMessageBo val = record.value();
            MessageDto messageDto = new MessageDto();
            messageDto.setType(val.getType());
            messageDto.setUniKey(val.getUniKey());
            messageDto.setData(val.getData());
            log.debug("send kafka msg:" + JSON.toJSONString(messageDto));
            messageDistributor.doSend(messageDto);
        }
    }

}
